package cn.admin.remoting.serialize.impl;

import cn.admin.remoting.Connection;
import cn.admin.remoting.impl.ConnectionRequest;
import cn.admin.remoting.impl.ConnectionResponse;
import cn.admin.remoting.impl.DefaultConnection;
import cn.admin.remoting.util.LoggerInit;
import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoFutureListener;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolDecoderAdapter;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;

import java.io.ByteArrayInputStream;
import java.io.IOException;

public class RemotingProtocolDecoder extends ProtocolDecoderAdapter {

    private static final String INVALID_PROTOCOL_HEADER_PREFIX = "非法的报文头字段: ";
    private static final String BUFFER = CumulativeProtocolDecoder.class.getName() + ".Buffer";
    private ConnRequestSerialization reqSerialize = new ConnRequestSerialization();
    private ConnResponseSerialization rspSerialize = new ConnResponseSerialization();

    @Override
    public final void decode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception {
        boolean useSessionBuffer = true;
        ByteBuffer buf = (ByteBuffer) session.getAttribute(BUFFER);
        if (buf != null) {
            buf.put(in);
            buf.flip();
        } else {
            buf = in;
            useSessionBuffer = false;
        }

        boolean keepRemaining = true;

        try {
            boolean decoded = true;

            while (decoded && buf.hasRemaining()) {
                int oldPos = buf.position();

                try {
                    decoded = this.doDecode(session,buf,out);
                    if (decoded) {
                        if (buf.position() == oldPos) {
                            throw new IllegalStateException("doDecode() can't return true when " +
                                    "buffer is not consumed.");
                        }
                    } else {
                        buf.position(oldPos);
                    }
                } catch (IOException e) {
                    if (e.getMessage().contains(INVALID_PROTOCOL_HEADER_PREFIX)) {
                        keepRemaining = false;
                        break;
                    }
                }
            }
        } finally {
            if (buf.hasRemaining() && keepRemaining) {
                if (useSessionBuffer) {
                    buf.compact();
                } else {
                    this.storeRemainingInSession(buf, session);
                }
            } else if (useSessionBuffer) {
                this.removeSessionBuffer(session);
            }
        }
    }

    protected boolean doDecode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out) throws Exception {
        int originPos = in.position();
        if (in.remaining() < 14) {
            return false;
        } else {
            byte protocolVer = in.get();
            if (protocolVer != 13) {
                IOException ioe = new IOException(INVALID_PROTOCOL_HEADER_PREFIX + protocolVer +
                        "," + session);
                LoggerInit.LOGGER.warn(ioe.getMessage(), ioe);
                throw ioe;
            } else {
                byte requestFlag = in.get();
                byte protocol = in.get();
                in.get();
                in.get();
                int conLength = in.getInt();
                int appClassNameLength = in.get();
                int appLength = in.getInt();
                int expectedBodyLength = conLength + appClassNameLength + appLength;
                if (in.remaining() < expectedBodyLength) {
                    in.position(originPos);
                    return false;
                } else {
                    in.position(originPos + 14 + expectedBodyLength);
                    ByteArrayInputStream is = new ByteArrayInputStream(in.array(), originPos,
                            14 + expectedBodyLength);
                    ConnectionRequest request;
                    if (0 == requestFlag) {
                        request = null;

                        try {
                            request = this.reqSerialize.deserialize(is, protocol);
                            out.write(request);
                        } catch (DeserializeException e) {
                            request = e.getRequest();
                            if (null != request) {
                                ConnectionResponse resp = ConnectionResponse.getResponse(request);
                                resp.setResult(1);
                                resp.setErrorMsg("反序列化业务请求时出现错误:" + e.getMessage());
                                session.write(resp);
                            }

                            throw e;
                        }
                    } else {
                        request = null;

                        ConnectionResponse response;
                        try {
                            response = this.rspSerialize.deserialize(is, protocol, session);
                            out.write(response);
                        } catch (DeserializeException e) {
                            response = null != e.getResponse() ? e.getResponse() :
                                    new ConnectionResponse();
                            response.setResult(5);
                            response.setThrowable(e);
                            Connection conn = DefaultConnection.findBySession(session);
                            conn.getClient().putResponse(response);
                            throw e;
                        }
                    }

                    return true;
                }
            }
        }
    }

    @Override
    public void dispose(IoSession session) throws Exception {
        this.removeSessionBuffer(session);
    }

    private void removeSessionBuffer(IoSession session) {
        ByteBuffer buf = (ByteBuffer) session.removeAttribute(BUFFER);
        if (buf != null) {
            buf.release();
        }
    }

    private void storeRemainingInSession(ByteBuffer buf, IoSession session) {
        ByteBuffer remainingBuf = ByteBuffer.allocate(buf.capacity());
        remainingBuf.setAutoExpand(true);
        remainingBuf.order(buf.order());
        remainingBuf.put(buf);
        session.setAttribute(BUFFER, remainingBuf);
    }
}
